package com.pxene.mr4hbase;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.util.Arrays;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.wltea.analyzer.core.IKSegmenter;
import org.wltea.analyzer.core.Lexeme;

public class ReadHBaseMapper extends TableMapper<Text, Text> {
 
    private Text resultKey = new Text();
    private Text resultVal = new Text();
 
    @Override
    protected void map(ImmutableBytesWritable key, Result value, Context context)
            throws IOException, InterruptedException {
        StringBuilder sb = new StringBuilder();
        // 通过cell获取表中的列值
        for (Cell cell : value.listCells()) {
        	if("article_content".equals(new String(CellUtil.cloneQualifier(cell))) || "article_title".equals(new String(CellUtil.cloneQualifier(cell)))){
        		String val = Bytes.toString(Arrays.copyOfRange(cell.getValueArray(), cell.getValueOffset(),
        				cell.getValueOffset() + cell.getValueLength()));
        		sb.append(val).append("|");
        	}
        }
        // 写入HDFS
//        resultKey.set(Bytes.toString(value.getRow()));
        resultKey.set("1");
        resultVal.set(IKAnalysis(sb.deleteCharAt(sb.length()-1).toString()));
        context.write(resultKey, resultVal);
    }
    
	public static String IKAnalysis(String str) {
		StringBuffer sb = new StringBuffer();
		try {
			// InputStream in = new FileInputStream(str);//
			byte[] bt = str.getBytes();// str
			InputStream ip = new ByteArrayInputStream(bt);
			Reader read = new InputStreamReader(ip);
			IKSegmenter iks = new IKSegmenter(read, true);
			Lexeme t;
			while ((t = iks.next()) != null) {
				sb.append(t.getLexemeText() + " ");
			}
			sb.delete(sb.length() - 1, sb.length());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		System.out.println(sb.toString());
		return sb.toString();
	}
 
}